package com.spark.sql

import org.apache.spark.SparkConf
import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.sql._

/**
  *
  * @author Pop
  * @date 2022/7/20 22:48
  */
object Spark04_SparkSql_UDAF1 {
  def main(args: Array[String]): Unit = {

    // 创建SparkSql的运行环境
    val sparkSql: SparkConf = new SparkConf().setMaster("local[*]").setAppName("sparkSql")
    val session: SparkSession = SparkSession.builder().config(sparkSql).getOrCreate()
    // 执行逻辑操作
      val df: DataFrame = session.read.json("datas/user.json")
    df.createOrReplaceTempView("user")

    // 希望如果你希望往结果定义一个前缀，可以使用自定义函数
    session.udf.register("ageA",functions.udaf(new MyAvgUDAF()))

    session.sql("select ageA(age) from user").show()

//      df.show
    // DataFrame =》 Sql
//    DataFrame2Sql(df,session)

  session.close()
  }

  // 定义一个缓冲区数据结构
  case class Buff(var total:Long,var count:Long)
  /**
    * 自定义聚合函数类，计算年龄的平均值
    *
    *  上一个例子，由于表结构是只有行和列的概念
    *  并没有类型，所以我们只能用索引位置来获得输入和缓冲区的数据
    *  但是很容易写错，这也是 UserDefinedAggregateFunction 被废弃的原因
    *  我们可以用这种，含有数据结构的方式来实现，自定义的聚合函数
    *
    *  引入 org.apache.spark.sql.expressions.Aggregator 定义泛型
    *  IN : 输入的数据类型 Long
    *  BUF: 缓冲区数据结构
    *  OUT：输入的数据类型 Long
    */
  class MyAvgUDAF extends Aggregator[Long,Buff,Long] {
    /**
      * 初始值或者零值
      *
      * 缓冲区的初始化
      * @return
      */
    override def zero: Buff = {
      Buff(0L,0L)
    }

    /**
      * 根据输入的数值更新缓冲区
      * @param b
      * @param a
      * @return
      */
    override def reduce(b: Buff, a: Long): Buff = {
      b.total = b.total+a
      b.count = b.count+1
      b
    }
    // 合并
    override def merge(b1: Buff, b2: Buff): Buff = {
      b1.total = b1.total + b2.total
      b1.count = b1.count + b2.count
      b1
    }
    //计算结果
    override def finish(reduction: Buff): Long = {
      reduction.total/reduction.count
    }
    //序列化编解码，用于网络传输
    override def bufferEncoder: Encoder[Buff] = Encoders.product

    override def outputEncoder: Encoder[Long] = Encoders.scalaLong
  }
}
